package com.demo.webflux.mongotx;

import java.time.Duration;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import com.demo.webflux.domain.Wallet;

import reactor.core.publisher.Mono;

@Service
public class TransactionWalletService {

	@Autowired
	private ReactiveMongoTemplate mongoTemplate;

	public Mono<TxResult> transferMoney(Mono<String> fromOwner, Mono<String> toOwner, Mono<Integer> requestAmount){
		return Mono.zip(fromOwner, toOwner, requestAmount)
					.flatMap(tuple3 -> {
						String from = tuple3.getT1();
						String to = tuple3.getT2();
						int amount = tuple3.getT3();
						return doTransferMoney(from, to, amount)
									.retryBackoff(20, Duration.ofMillis(1), Duration.ofMillis(500), 0.1)
									.onErrorReturn(TxResult.TX_CONFLICT);
					});
	}

	// 进行实际的转账操作
	private Mono<TxResult> doTransferMoney (String from, String to, Integer amount){
		// 开启事务，获得了ReactiveMongoOperations实例session
		return mongoTemplate.inTransaction().execute(session -> session
						.findOne(queryFroOwner(from), Wallet.class) // 先搜索汇款人的钱包
						.flatMap(fromWallet -> session
								.findOne(queryFroOwner(to), Wallet.class) // 再搜索收款人的钱包
								.flatMap(toWallet -> {
									if(fromWallet.hasEnoughFunds(amount)) { // 在解析了两个钱包后，检查汇款人是否有足够的资金
										fromWallet.withdraw(amount); // 然后从汇款人的钱包中提取正确的金额
										toWallet.deposit(amount); // 并将相同的资金存入收款人的账户，此时，尚未更新数据库
										return session.save(fromWallet) // 保存汇款人更新后的钱包
												.then(session.save(toWallet)) // 再保存收款人更新后的钱包
												.then(Mono.just(TxResult.SUCCESS)); // 如果数据库没有拒绝更改，就返回SUCCESS，并自动提交事务
									}else {
										return Mono.just(TxResult.NOT_ENOUGH_FUNDS); // 如果汇款人没有足够的资金，返回NOT_ENOUGH_FUNDS
									}
								})))
				.onErrorResume(e -> Mono.error(new RuntimeException("Confilict"))) // 如果在数据库通讯时出现任何错误，我们会传播onError信号，会触发重试逻辑
				.last();
	}

	private Query queryFroOwner(String owner) {
		return Query.query(new Criteria("owner").is(owner));
	}

	public static enum TxResult{
		SUCCESS,
		NOT_ENOUGH_FUNDS,
		TX_CONFLICT,
		;
	}

}
